with problem_bill_base as (
select
     t.id
    ,t.waybill_no
    ,max(t.create_time) as  create_time
    ,max(t.scan_time) as scan_time
    ,t.registration_network_name  --登记网点
    ,t.registration_network_code
    ,t.proble_type_subject_name   --一级类型
    ,t.proble_type_subject_code
    ,t.second_level_type_code    --二级类型
    ,t.second_level_type_name
    ,t.receive_network_name  --接受网点
    ,t.receive_network_code  --接受网点code
from jms_dwd.dwd_sqs_registration_problem_piece_base_dt  t
where t.dt>='{{ execution_date | date_add(-15) | cst_ds }}'
      and t.dt<='{{ execution_date  | cst_ds }}'
      and t.create_time >='{{ execution_date | date_add(-1) | cst_ds }}'
      and t.proble_type_subject_code='29'
      and t.second_level_type_code='29a'
group by
     t.id
    ,t.waybill_no
    ,t.registration_network_name  --登记网点
    ,t.registration_network_code
    ,t.proble_type_subject_name   --一级类型
    ,t.proble_type_subject_code
    ,t.second_level_type_code    --二级类型
    ,t.second_level_type_name
    ,t.receive_network_name
    ,t.receive_network_code

)
--经停表
,tmp_shipment_stop_base as (
    select
      t1.shipment_no
     ,t1.network_code
     ,t1.network_name
     ,t1.actual_arrival_time   --到车时间
     ,t1.planned_arrival_time --计划倒车时间
     ,lag(t1.planned_departure_time)over(partition by t1.shipment_no order by t1.sort_num) as planned_departure_time  --计划发车时间
    from jms_dwd.dwd_tmsnew_shipment_stop_union_base_dt t1
    where t1.dt>='{{ execution_date | date_add(-20) | cst_ds }}'
          and t1.dt<='{{ execution_date  | cst_ds }}'
)

--发件表，获取到达时间
,tmp_send_bill as (
    select
          t.waybill_no
         ,t.scan_time
         ,t.input_site as scan_site
         ,t.scan_site_code
         ,t.next_station
         ,t.next_station_code
         ,1 as scan_type
         ,t1.actual_arrival_time
         ,unix_timestamp(t.scan_time,'yyyy-MM-dd HH:mm:ss') as scan_time_unix
         ,abs(unix_timestamp(t1.planned_arrival_time,'yyyy-MM-dd HH:mm:ss') - unix_timestamp(t1.planned_departure_time,'yyyy-MM-dd HH:mm:ss')) as stop_time
    from jms_dwd.dwd_tab_barscan_send_base_dt t
    left join  tmp_shipment_stop_base t1  on t.shipment_no=t1.shipment_no  and t.next_station_code=t1.network_code
    where t.dt>='{{ execution_date | date_add(-20) | cst_ds }}'
          and t.dt<='{{ execution_date  | cst_ds }}'
          and t.scan_time>='{{ execution_date | date_add(-15) | cst_ds }}'
          and t.scan_site_code is not null
          and t.waybill_no is not null
)

--扫描记录表
,tmp_problem_bill1 as (
    select * from (
                               --发件
                                select
                                      t.waybill_no
                                     ,t.scan_time
                                     ,t.scan_site
                                     ,t.scan_site_code
                                     ,t.next_station
                                     ,t.next_station_code
                                     ,1 as scan_type
                                     ,t1.scan_time as problem_scan_time
                                from  problem_bill_base t1
                                join tmp_send_bill t on t1.waybill_no=t.waybill_no

                                union all
                                --建包
                                select
                                      t.waybill_no
                                     ,t.scan_time
                                     ,t.scan_site
                                     ,t.scan_site_code
                                     ,null as next_station
                                     ,null as next_station_code
                                     ,2 as scan_type
                                     ,t1.scan_time as problem_scan_time
                                from  problem_bill_base t1
                                join jms_dwd.dwd_tab_barscan_bagging_base_dt t on t1.waybill_no=t.waybill_no
                                where t.dt>='{{ execution_date | date_add(-20) | cst_ds }}'
                                  and t.dt<='{{ execution_date  | cst_ds }}'
                                       and t.scan_time>='{{ execution_date | date_add(-15) | cst_ds }}'
                                union all
                                --拦截
                                select
                                      t.waybill_no
                                     ,nvl(t.intercept_time,t.scan_time) as scan_time
                                     ,nvl(t.scan_network_name,intercept_network_name )as scan_site
                                     ,nvl(t.scan_network_code,intercept_network_code )as scan_site_code
                                     ,null as next_station
                                     ,null as next_station_code
                                     ,3 as scan_type
                                     ,t1.scan_time as problem_scan_time
                                from problem_bill_base t1
                                join jms_dwd.dwd_yl_oms_interceptorpiece_base t on t1.waybill_no=t.waybill_no
                                where  t.scan_time>='{{ execution_date | date_add(-15) | cst_ds }}'
                                    and t.scan_time<='{{ execution_date  | cst_ds }}'
                                union all
                                --退件，转寄件
                                select
                                      t.waybill_no
                                     ,nvl(t.print_time,t.apply_time) as scan_time
                                     ,nvl(t.print_network_name,t.apply_network_name) as scan_site
                                     ,nvl(t.print_network_code,t.apply_network_code) as scan_site_code
                                     ,null as next_station
                                     ,null as next_station_code
                                     ,4 as scan_type
                                     ,t1.scan_time as problem_scan_time

                                from  problem_bill_base t1
                                join jms_dwd.dwd_tab_reback_transfer_express_base t on t1.waybill_no=t.waybill_no
                                where t.apply_time>='{{ execution_date | date_add(-15) | cst_ds }}'
                                     and t.apply_time<='{{ execution_date  | cst_ds }}'
                                union all
                                --问题件
                                select
                                      t.waybill_no
                                     ,t.scan_time
                                     ,t.registration_network_name as scan_site
                                     ,t.registration_network_code as scan_site_code
                                     ,null as next_station
                                     ,null as next_station_code
                                     ,5 as scan_type
                                     ,t1.scan_time as problem_scan_time

                                from  problem_bill_base t1
                                join jms_dwd.dwd_sqs_registration_problem_piece_base_dt t on t1.waybill_no=t.waybill_no
                                where t.dt>='{{ execution_date | date_add(-20) | cst_ds }}'
                                      and t.dt<='{{ execution_date  | cst_ds }}'
                                      and t.scan_time>='{{ execution_date | date_add(-15) | cst_ds }}'
                                      and t.proble_type_subject_code<>'29'
                                      and t.proble_type_subject_code<>'50'
                                union all
                                --派件，出仓扫描
                                select
                                      t.waybill_no
                                     ,t.scan_time
                                     ,t.scan_site
                                     ,t.scan_site_code
                                     ,null as next_station
                                     ,null as next_station_code
                                     ,6 as scan_type
                                     ,t1.scan_time as problem_scan_time
                                from problem_bill_base t1
                                join jms_dwd.dwd_tab_barscan_deliver_base_dt t on t1.waybill_no=t.waybill_no
                                where t.dt>='{{ execution_date | date_add(-20) | cst_ds }}'
                                  and t.dt<='{{ execution_date  | cst_ds }}'
                                      and t.scan_time>='{{ execution_date | date_add(-15) | cst_ds }}'

                                union all
                                --代收，入库扫描
                                select
                                      t.waybill_no
                                     ,t.scan_time
                                     ,t.scan_site
                                     ,t.scan_site_code
                                     ,null as next_station
                                     ,null as next_station_code
                                     ,7 as scan_type
                                     ,t1.scan_time as problem_scan_time

                                from problem_bill_base t1
                                join jms_dwd.dwd_tab_barscan_collect_base_dt t on t1.waybill_no=t.waybill_no
                                where t.dt>='{{ execution_date | date_add(-20) | cst_ds }}'
                                  and t.dt<='{{ execution_date  | cst_ds }}'
                                  and t.scan_time>='{{ execution_date | date_add(-15) | cst_ds }}'
                                  and t.scan_type='入库扫描'
                )     t

)


--问题一，二,三,四，基础表
,problem_bill_base1 as (
    select
         t.id
        ,t.waybill_no
        ,t.create_time   --登记时间
        ,t.scan_time       --扫描时间
        ,to_date(t.create_time) as create_time_f --登记时间格式化
        ,t.registration_network_name  --登记网点
        ,t.registration_network_code
        ,t.proble_type_subject_name   --一级类型
        ,t.proble_type_subject_code
        ,t.second_level_type_code    --二级类型
        ,t.second_level_type_name
        ,t1.scan_site_code
        ,t1.scan_time  as send_scan_time   --发件扫描时间
        ,t2.scan_time  as problem_scan_time --问题件登记网点扫描时间,
        ,t2.scan_type
        ,t2.scan_site_code  as up_site_code
        --t1发车时间，t2问题件扫描时间 看问题件发车时间和登记时间之间有没有本站扫描
        ,max(case when   t.scan_time > t1.scan_time then t1.scan_time else null end )over(partition by t.id) as max_site_time --最近的发车扫描时间

        --,case when (t2.scan_type<> 6 and t2.scan_type <> 7) and t2.scan_time > t1.scan_time   and t2.scan_time< t.scan_time then 1 else 0 end  as one_flag

        --t1发车时间，t2问题件扫描时间，看登记时间之前的发车扫描，登记时间是不是小于时间发车时间+规定运输时间
        ,case when  t.scan_time > t1.scan_time   and t.scan_time < from_unixtime(t1.scan_time_unix+t1.stop_time,'yyyy-MM-dd HH:mm:ss') then 1 else 0 end  as two_flag

         --t1发车时间，t2问题件扫描时间，,t登记时间
        ,max(case when  t.scan_time > t2.scan_time   then t2.scan_time else null end )over(partition by t.id) as up_site_scan_time --最近的上一站扫描时间

         --t1发车时间，t2问题件扫描时间，,t登记时间, 看距离登记时间之后最近的扫描记录
        ,min(case when t.receive_network_code=t1.scan_site_code and t.scan_time < t2.scan_time and (t2.scan_type<> 6 and t2.scan_type <> 7) then t2.scan_time else null end )over(partition by t.id) as down_site_time

    from problem_bill_base t
    left join tmp_send_bill t1 on t.waybill_no=t1.waybill_no and t.registration_network_code=t1.next_station_code  and t.receive_network_code=t1.scan_site_code
    left join tmp_problem_bill1 t2 on t.waybill_no=t2.waybill_no and t.registration_network_code=t2.scan_site_code
)


,problem_bill_base2 as  (
    select
         t.id
        ,t.waybill_no
        ,t.create_time   --登记时间
        ,t.scan_time       --扫描时间
        ,t.create_time_f --登记时间格式化
        ,t.registration_network_name  --登记网点
        ,t.registration_network_code
        ,t.proble_type_subject_name   --一级类型
        ,t.proble_type_subject_code
        ,t.second_level_type_code    --二级类型
        ,t.second_level_type_name
        ,t.scan_site_code
        ,t.send_scan_time   --发件扫描时间
        ,t.problem_scan_time --登记网点扫描时间,
        ,t.max_site_time
        ,case when (t.scan_type<> 6 and t.scan_type <> 7) and t.problem_scan_time > t.max_site_time   and t.problem_scan_time< t.scan_time then 1 else 0 end one_flag
        ,t.two_flag
        ,case when t.up_site_scan_time=t.problem_scan_time and t.problem_scan_time> t.max_site_time then t.up_site_code end as up_site_code
        ,t.down_site_time
        ,case when t.up_site_scan_time=t.problem_scan_time and t.problem_scan_time> t.max_site_time then
            locate(t.registration_network_code, tt.in_distribution_code_new)+locate(t.up_site_code, tt1.in_distribution_code_new)  else 0 end as in_count
        ,case when t.up_site_scan_time=t.problem_scan_time and t.problem_scan_time> t.max_site_time then
            locate(t.registration_network_code, tt.out_distribution_code_new)+locate(t.up_site_code, tt1.out_distribution_code_new) else 0 end as out_count
    from problem_bill_base1 t
    left join jms_dim.dim_network_whole_massage  tt on t.up_site_code=tt.code
    left join jms_dim.dim_network_whole_massage  tt1 on t.registration_network_code=tt1.code
)
--问题四
,tmp_banci as (
     select
         t2.scan_site_code
        ,t2.next_station_code
        ,t2.actual_arrival_time
     from(
           select
              t1.scan_site_code
            ,t1.next_station_code
           from problem_bill_base t
           join tmp_send_bill t1 on t.waybill_no=t1.waybill_no and t.registration_network_code=t1.next_station_code
           where t1.scan_site_code is not null and t1.next_station_code is not null
           group by  t1.scan_site_code
            ,t1.next_station_code
       ) t1
       left join tmp_send_bill t2 on t1.scan_site_code=t2.scan_site_code  and t1.next_station_code=t2.next_station_code
       where    t2.scan_site_code  is not null
              and t2.next_station_code is not null
              and t2.actual_arrival_time is not null
       group by
         t2.scan_site_code
        ,t2.next_station_code
        ,t2.actual_arrival_time
)



insert overwrite table jms_dm.dm_send_no_arrival_detail_dt partition ( dt )

select
 t.date_time
,t.id
,t.waybill_no
,t.create_time
,t.registration_network_name
,t.registration_network_code
,case when t.one_flag=1 then 1
      when t.two_flag=1 then 1
      when t.three_flag=1 then 1
      when t.four_flag=0 then 1
      else 0
  end  is_error

,case when t.one_flag=1 then 1
      when t.two_flag=1 then 2
      when t.three_flag=1 then 3
      when t.four_flag=0 then 4
      else 0
  end  is_error_type
,t.date_time as dt
from (
    select
     t.create_time_f as date_time
    ,t.id
    ,t.waybill_no
    ,max(t.create_time) as create_time
    ,max(t.registration_network_name)  as registration_network_name--登记网点
    ,t.registration_network_code as registration_network_code
    ,max(t.one_flag) as one_flag
    ,max(t.two_flag) as two_flag
    ,max(case when t.up_site_code is not null and (in_count+out_count)>=1 then 1 else 0 end ) as  three_flag
    ,max(case when  (t2.actual_arrival_time > t.scan_time and t2.actual_arrival_time < t.down_site_time) or down_site_time is null  or t2.actual_arrival_time is null then 1 else 0 end  ) as  four_flag
    ,t.create_time_f as dt
    from
        problem_bill_base2 t
    left join tmp_banci t2 on t.scan_site_code=t2.scan_site_code and t.registration_network_code=t2.next_station_code
    where create_time_f='{{ execution_date | date_add(-1) | cst_ds }}'
    group by
     t.id
    ,t.waybill_no
    ,t.create_time_f
    ,t.registration_network_code
) t

 distribute by pmod(hash(rand()), 3);

